package com.atguigu.fink.chapter01.window;

import com.atguigu.fink.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Author lzc
 * @Date 2022/11/28 15:09
 */
public class Flink03_TVF_1 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> stream = env
            .fromElements(
                new WaterSensor("s1", 1000L, 10),
                new WaterSensor("s1", 2000L, 10),
                new WaterSensor("s1", 3000L, 20),
                new WaterSensor("s1", 4000L, 30),
                new WaterSensor("s1", 7000L, 40),
                new WaterSensor("s1", 9000L, 40),
                new WaterSensor("s1", 12000L, 40),
                new WaterSensor("s1", 17000L, 40),
                new WaterSensor("s1", 22000L, 40),
                new WaterSensor("s1", 24000L, 40),
                new WaterSensor("s1", 30000L, 50),
                new WaterSensor("s1", 31000L, 50)
            )
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((ws, ts) -> ws.getTs())
            
            );
        
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        Table table = tEnv.fromDataStream(stream, $("id"), $("ts").rowtime(), $("vc"));
        tEnv.createTemporaryView("sensor", table);
        
        tEnv.sqlQuery("select " +
                          "window_start, " +
                          "window_end, " +
                          "id, " +
                          "sum(vc) vc_sum " +
                          "from table( cumulate( table sensor, descriptor(ts), interval '5' second,interval '20' second) )" +
                          "group by id, window_start, window_end")
            .execute()
            .print();
        
        
    }
}
/*
每个 1h 计算一下当天的 pv 的值
    0-1
    0-2
    0-3
    ...

如果用流来计算:
    思路1:
        定义窗口长度是 1h 的滚动窗口
            聚和的时候不清空状态,当到第二天的第一个窗口才清空
            
    思路 2:
        定义一个长度为 24h 的滚动窗口
        
        自定义触发器 Trigger, 每隔一小时触发一次计算, 但是不关闭窗口
        
如果用 sql:
    用 commulate 窗口, 直接实现这种功能
        

 */